#include <amqp.h>
#include <amqp_framing.h>
#include "rabbit.h"

static const string MODULE = "RABBIT";

Consumer::Consumer()
{
    alive = false;
    thread = nullptr;
}

Consumer::~Consumer()
{
    stop();
}

int Consumer::start(const Config &conf)
{
    config = conf;
    alive = true;
    thread = new std::thread(entry, this);
    return 0;
}

void Consumer::stop()
{
    if (alive)
    {
        alive = false;
        thread->join();
    }
}

void Consumer::main_entry()
{
    amqp_connection_state_t conn = NULL;
    amqp_channel_t channel = 1;
    amqp_rpc_reply_t ret;

    while (alive)
    {
        conn = amqp_new_connection();
        if (!conn)
        {
            logger.log_notice(MODULE, "fail to alloc connection");
            continue;
        }

        int sock = amqp_open_socket(config.server_host.c_str(), config.server_port);
        if (sock < 0)
        {
            logger.log_notice(MODULE, "fail to connect server(%s:%d)", config.server_host.c_str(), config.server_port);
            amqp_destroy_connection(conn);
			conn = NULL;
            util::sleep(1000);
            continue;
        }

		//If an input or output function blocks for this period of time, and data has been sent or received, the return value of
		//that function will be the amount of data transferred; if no data has been transferred and the timeout has been reached
		//then -1 is returned with errno set to EAGAIN or EWOULDBLOCK just as if the socket was specified to be nonblocking.

		struct timeval val;
		val.tv_sec = 1;
		val.tv_usec = 0;
		setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char *)&val, sizeof(struct timeval));

        amqp_set_sockfd(conn, sock);

		ret = amqp_login(conn, config.vhost.c_str(), 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, config.user_name.c_str(), config.password.c_str());
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
        {
            logger.log_notice(MODULE, "fail to login rabbitmq: %d", ret.reply_type);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
        }

        amqp_channel_open(conn, channel);
        ret = amqp_get_rpc_reply(conn);
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
        {
            logger.log_notice(MODULE, "fail to open channel: %d", ret.reply_type);
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
        }

		amqp_bytes_t amqp_exchange_name = amqp_cstring_bytes(config.exchange_name.c_str());
		amqp_bytes_t amqp_exchange_type = amqp_cstring_bytes(config.exchange_type.c_str());
		amqp_bytes_t amqp_queue_name = amqp_cstring_bytes(config.queue_name.c_str());
		amqp_bytes_t amqp_binding_key = amqp_cstring_bytes(config.binding_key.c_str());

        //passive: If set, the server will reply with Declare-Ok if the exchange already exists
        //  with the same name, and raise an error if not. The client can use this to check
        //  whether an exchange exists without modifying the server state.

        //durable: If set when creating a new exchange, the exchange will be marked as durable.
        //  Durable exchanges remain active when a server restarts. Non-durable exchanges
        //  (transient exchanges) are purged if/when a server restarts.

		amqp_exchange_declare(conn, channel, amqp_exchange_name, amqp_exchange_type, 0, 0, amqp_empty_table);
		ret = amqp_get_rpc_reply(conn);
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
		{
            logger.log_notice(MODULE, "fail to declare exchange: %d", ret.reply_type);
            amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
		}

		//passive: If set, the server will reply with Declare-Ok if the queue already exists
		//  with the same name, and raise an error if not. The client can use this to check
		//  whether a queue exists without modifying the server state.

		//durable: If set when creating a new queue, the queue will be marked as durable.
		//  Durable queues remain active when a server restarts. Non-durable queues
		//  (transient queues) are purged if/when a server restarts.

		//exclusive: Exclusive queues may only be accessed by the current connection,
		//  and are deleted when that connection closes. Passive declaration of an exclusive
		//  queue by other connections are not allowed.

		//auto_delete: If set, the queue is deleted when all consumers have finished using it.
		//  The last consumer can be cancelled either explicitly or because its channel is closed.
		//  If there was no consumer ever on the queue, it won't be deleted.

        amqp_queue_declare_ok_t *rok = amqp_queue_declare(conn, channel, amqp_queue_name, 0, 0, 0, 1, amqp_empty_table);
        ret = amqp_get_rpc_reply(conn);
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
        {
            logger.log_notice(MODULE, "fail to declare queue: %d", ret.reply_type);
            amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
        }

		config.queue_name.assign((const char *)rok->queue.bytes, rok->queue.len);
		amqp_queue_name = amqp_cstring_bytes(config.queue_name.c_str());

		amqp_queue_bind(conn, channel, amqp_queue_name, amqp_exchange_name, amqp_binding_key, amqp_empty_table);
		ret = amqp_get_rpc_reply(conn);
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
        {
            logger.log_notice(MODULE, "fail to bind queue: %d", ret.reply_type);
            amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
        }

		//no_local:
		//no_ack:
		//exclusive: Request exclusive consumer access, meaning only this consumer can access the queue.

        amqp_basic_consume(conn, channel, amqp_queue_name, amqp_empty_bytes, 0, 1, 0, amqp_empty_table);
        ret = amqp_get_rpc_reply(conn);
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
        {
            logger.log_notice(MODULE, "fail to basic consume: %d", ret.reply_type);
            amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
        }

		logger.log_notice(MODULE, "consumer connected to server");

        amqp_frame_t frame;
        uint64_t body_target;
        uint64_t body_recved;

        while (alive)
        {
            amqp_maybe_release_buffers(conn);

            if (amqp_simple_wait_frame(conn, &frame) < 0)
            {
#ifdef WIN32
				if (WSAGetLastError() == WSAETIMEDOUT)
						continue;
#else
				if (errno == EAGAIN)
						continue;
#endif
                logger.log_notice(MODULE, "fail to wait frame");
                break;
            }

            if (frame.frame_type != AMQP_FRAME_METHOD)
            {
                logger.log_notice(MODULE, "frame type unexpected");
                continue;
            }

            if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)
    	    {
                logger.log_notice(MODULE, "frame method id unexpected");
                continue;
            }

            if (amqp_simple_wait_frame(conn, &frame) < 0)
    	    {
                logger.log_notice(MODULE, "fail to wait frame");
                break;
            }

            if (frame.frame_type != AMQP_FRAME_HEADER)
    	    {
                logger.log_notice(MODULE, "frame type unexpected");
                continue;
            }

            body_target = frame.payload.properties.body_size;
            body_recved = 0;

            string buf = "";

            while (body_recved < body_target)
            {
                if (amqp_simple_wait_frame(conn, &frame) < 0)
    	        {
                    logger.log_notice(MODULE, "fail to wait frame");
                    break;
                }

                if (frame.frame_type != AMQP_FRAME_BODY)
                {
                    logger.log_notice(MODULE, "frame type unexpected");
                    break;
                }

                body_recved += frame.payload.body_fragment.len;
                buf.append((const char *)frame.payload.body_fragment.bytes, frame.payload.body_fragment.len);
            }

            if (body_recved != body_target)
            {
                logger.log_notice(MODULE, "frame length unexpected");
                break;
            }

            logger.log_debug(MODULE, "rabbit consume message: %s", buf.c_str());
            config.received(std::move(buf));
        }

		logger.log_notice(MODULE, "consumer disconn from server");
		amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(conn);
		conn = NULL;
    }

    if (conn)
    {
        logger.log_notice(MODULE, "consumer disconn and exit");
        amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
        amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
        amqp_destroy_connection(conn);
        conn = NULL;
    }
}

void Consumer::entry(void *arg)
{
    Consumer *consumer = (Consumer *)arg;
    consumer->main_entry();
}

Publisher::Publisher()
{
    alive = false;
    thread = nullptr;
}

Publisher::~Publisher()
{
    stop();
}

int Publisher::start(const Config &conf)
{
    config = conf;
    alive = true;
    thread = new std::thread(entry, this);
	return 0;
}

void Publisher::stop()
{
    if (alive)
    {
        alive = false;
        thread->join();
    }
}

void Publisher::send(const string &routing_key, const string &content)
{
    Message *msg = new Message();
    msg->routing_key = routing_key;
    msg->content = content;
    cache_queue.push(msg);
}

void Publisher::main_entry()
{
    amqp_connection_state_t conn = NULL;
    amqp_channel_t channel = 1;
    amqp_rpc_reply_t ret;

    while (alive)
    {
        amqp_connection_state_t conn = amqp_new_connection();
        if (!conn)
        {
            logger.log_notice(MODULE, "fail to alloc connection");
            continue;
        }

        int sock = amqp_open_socket(config.server_host.c_str(), config.server_port);
        if (sock < 0)
        {
            logger.log_notice(MODULE, "fail to connect server(%s:%d)", config.server_host.c_str(), config.server_port);
            amqp_destroy_connection(conn);
            conn = NULL;
            util::sleep(1000);
            continue;
        }

        amqp_set_sockfd(conn, sock);

		ret = amqp_login(conn, config.vhost.c_str(), 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, config.user_name.c_str(), config.password.c_str());
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
        {
            logger.log_notice(MODULE, "fail to login rabbitmq: %d", ret.reply_type);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
        }

        amqp_channel_open(conn, channel);
        ret = amqp_get_rpc_reply(conn);
        if (ret.reply_type != AMQP_RESPONSE_NORMAL)
        {
            logger.log_notice(MODULE, "fail to open channel: %d", ret.reply_type);
            amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
            conn = NULL;
            continue;
        }

		logger.log_notice(MODULE, "publisher connected to server");

        Message tmp;

        while (alive)
        {
            if (cache_queue.wait(tmp, 100))
            {
                continue;
            }

			amqp_bytes_t amqp_exchange_name = amqp_cstring_bytes(config.exchange_name.c_str());
            amqp_bytes_t amqp_routing_key = amqp_cstring_bytes(tmp.routing_key.c_str());
            amqp_bytes_t body_bytes = amqp_cstring_bytes(tmp.content.c_str());

			amqp_basic_properties_t props;
			props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
			props.content_type = amqp_cstring_bytes("text/plain");
			props.delivery_mode = 2; //persistent delivery mode

			//mandatory: This flag tells the server how to react if the message cannot be routed to a queue.
			//  If this flag is set, the server will return an unroutable message with a Return method.
			//  If this flag is zero, the server silently drops the message.

			//immediate: This flag tells the server how to react if the message cannot be routed to a queue
			//  consumer immediately. If this flag is set, the server will return an undeliverable message
			//  with a Return method. If this flag is zero, the server will queue the message, but with no
			//  guarantee that it will ever be consumed.

			int ret = amqp_basic_publish(conn, channel, amqp_exchange_name, amqp_routing_key, 0, 0, &props, body_bytes);
            if (ret < 0)
            {
                logger.log_notice(MODULE, "fail to publish message");
                break;
            }

			logger.log_debug(MODULE, "rabbit publish message: %s", tmp.content.c_str());
        }

		logger.log_notice(MODULE, "publisher disconn from server");
		amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(conn);
		conn = NULL;
    }

    if (conn)
    {
        logger.log_notice(MODULE, "publisher disconn and exit");
        amqp_channel_close(conn, channel, AMQP_REPLY_SUCCESS);
        amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
        amqp_destroy_connection(conn);
        conn = NULL;
    }
}

void Publisher::entry(void *arg)
{
    Publisher *publisher = (Publisher *)arg;
    publisher->main_entry();
}

